package common

import (
	"encoding/hex"
	"errors"
	"fmt"
	"strconv"
	"sync"
	"sync/atomic"
	"time"

	"github.com/allegro/bigcache"
	"github.com/golang/protobuf/proto"
	"hundsun.com/hsl/hschain/blockchain/impl/pubsub"
	"hundsun.com/hsl/hschain/common/cache"
	"hundsun.com/hsl/hschain/common/config"
	"hundsun.com/hsl/hschain/common/keystore"
	"hundsun.com/hsl/hschain/common/log/logging"
	"hundsun.com/hsl/hschain/common/types"
	"hundsun.com/hsl/hschain/common/util"
	"hundsun.com/hsl/hschain/common/util/pb"
	consensustypes "hundsun.com/hsl/hschain/consensus/types"
	"hundsun.com/hsl/hschain/module"
	"hundsun.com/hsl/hschain/protos"
	commonproto "hundsun.com/hsl/hschain/protos/common"
	protosconfig "hundsun.com/hsl/hschain/protos/config"
	"hundsun.com/hsl/hschain/protos/consensus/dpos"
	pbledger "hundsun.com/hsl/hschain/protos/ledger"
	"hundsun.com/hsl/hschain/protos/p2p"
	"hundsun.com/hsl/hschain/store"
	dbm "hundsun.com/hsl/hschain/store/db"
	"hundsun.com/hsl/hschain/store/mq"
)

// BaseClient 公共mqClient
type BaseClient struct {
	topicClient       mq.Client
	workerClient      mq.Client
	isClosed          int32
	currentBlock      *pbledger.Block
	muLock            sync.Mutex
	blockExecutedFunc func(block *pbledger.ExecutedBlock) error
	blockCheckedFunc  func(block *pbledger.Block) error
	blockReceivedFunc func(block *pbledger.Block) error
	blockAddReplyFunc func(block *pbledger.Block) error

	// 区块是否已同步：最近验证的块的序列号需要与序列号中心保持一致；
	blockSynchronized       *atomic.Value
	blockExecutedMessage    *cache.Cache
	lastBlockCheckedMessage *atomic.Value
	lastBlockSequence       *atomic.Value
	markedTxLocalCache      *bigcache.BigCache

	relaymsgRequestFunc  func(msg *p2p.P2PMsg) error
	relayMsgResponseFunc func(msg *p2p.P2PMsg) error

	// 区块产生流程相关
	producerConf                     *consensustypes.ProducerConfig
	shouldProceedProducingBlocksFunc func() bool
	beforeBlockCreatedFunc           func() (lastBlockNumber uint64, lastBlockHash protos.Hash)
	afterBlockCreatedFunc            func(block *pbledger.Block)

	blockSequenceServer *BlockSequenceServiceServer

	// 区块验证流程相关，后续将移动到blockchain
	checkerConf       *consensustypes.CheckerConfig
	checkedBlkQueue   chan *consensustypes.CheckedBlock
	maxRecvBlkSeq     *atomic.Value
	lastCheckedBlkSeq *atomic.Value   //上一个校验的区块序号
	lastAcceptedBlock *pbledger.Block //上一个验证通过的块

	// 存储账本数据句柄
	ledgerDB dbm.Database
	// 存储状态数据DB句柄
	stateDB dbm.Database

	// 系统配置项管理
	curParam *protosconfig.SysParam

	// 共识切换配置项管理
	moduleCfg *config.ModuleConfig
	curCon    ConsensusModule

	stepConMutex   sync.RWMutex
	stepParamMutex sync.RWMutex

	log logging.Log
}

// NewBaseClient 新建BaseClient
func NewBaseClient() *BaseClient {
	tClient, err := store.NewMqClient(mq.ModeTopic.String())
	if err != nil {
		panic(err)
	}

	wClient, err := store.NewMqClient(mq.ModeWorker.String())
	if err != nil {
		panic(err)
	}

	tClient.Sub(types.ConsensusModule, mq.ModeTopic.String())
	wClient.Sub(types.ConsensusModule, mq.ModeWorker.String())

	bc := &BaseClient{
		topicClient:             tClient,
		workerClient:            wClient,
		blockSynchronized:       &atomic.Value{},
		lastBlockCheckedMessage: &atomic.Value{},
		lastBlockSequence:       &atomic.Value{},
		ledgerDB:                store.GetLedgerDB(),
		stateDB:                 store.GetStateDB(),
		curParam:                &protosconfig.SysParam{},
		log:                     logging.NewLog(logging.Info, "baseclient", "client"),
	}
	bc.lastBlockCheckedMessage.Store(&commonproto.BlockCheckedMessage{})
	bc.lastBlockSequence.Store(uint64(0))

	return bc
}

// InitBlockSequenceServer ...
func (bc *BaseClient) InitBlockSequenceServer(ks keystore.KeyStore) error {
	if bc.blockSequenceServer != nil {
		return nil
	}
	checkBlockEnabled := bc.GetCurParam().CheckBlock.Enabled
	blockSeqSignerAddr := pb.GetGenesisTransactionPayload().GetSequenceCenter().GetSignerAddress()
	blockSequenceSignerAccount := ks.GetAccount(protos.Address(blockSeqSignerAddr))
	if blockSequenceSignerAccount == nil {
		return errors.New("block signer '" + blockSeqSignerAddr + "' not found in key store")
	}

	maxBlockSize := pb.GetGenesisTransactionPayload().GetBlockLimit().GetMaxBlockSizeBytes()

	serverConfig := &BlockSequenceServiceServerConfig{
		MaxBlockSize:      maxBlockSize,
		CheckBlockEnabled: checkBlockEnabled,
		RequestQueueSize:  0,
	}

	bc.blockSequenceServer = NewBlockSequenceServiceServer(serverConfig)
	bc.blockSequenceServer.sequenceSigner = blockSequenceSignerAccount
	bc.blockSequenceServer.baseclient = bc

	return bc.blockSequenceServer.Start(bc.log)
}

// GetSequenceServer 获取sequence server
func (bc *BaseClient) GetSequenceServer() *BlockSequenceServiceServer {
	return bc.blockSequenceServer
}

// SetSequenceServer 设置sequence server
func (bc *BaseClient) SetSequenceServer(server *BlockSequenceServiceServer) {
	bc.blockSequenceServer = server
}

// GetTopicClient 获取Topic类型Client
func (bc *BaseClient) GetTopicClient() mq.Client {
	return bc.topicClient
}

// GetWorkerClient 获取Worker类型Client
func (bc *BaseClient) GetWorkerClient() mq.Client {
	return bc.workerClient
}

// Wait wait for ready
func (bc *BaseClient) Wait() {}

// SetTopicClient 设置TopicClient
func (bc *BaseClient) SetTopicClient(c mq.Client) {
	bc.topicClient = c
}

// SetWorkerClient 设置WorkerClient
func (bc *BaseClient) SetWorkerClient(c mq.Client) {
	bc.workerClient = c
}

// GetLedgerDB 获取ledgerDB
func (bc *BaseClient) GetLedgerDB() dbm.Database {
	return bc.ledgerDB
}

// SetLedgerDB 设置ledgerDB
func (bc *BaseClient) SetLedgerDB(db dbm.Database) {
	bc.ledgerDB = db
}

// GetStateDB 获取stateDB
func (bc *BaseClient) GetStateDB() dbm.Database {
	return bc.stateDB
}

// SetCurrentCon set current consensus
func (bc *BaseClient) SetCurrentCon(con ConsensusModule) {
	bc.curCon = con
}

// GetCurrentCon get current consensus
func (bc *BaseClient) GetCurrentCon() ConsensusModule {
	return bc.curCon
}

// SetModuleCfg set module config
func (bc *BaseClient) SetModuleCfg(cfg *config.ModuleConfig) {
	bc.moduleCfg = cfg
}

// GetModuleCfg get module config
func (bc *BaseClient) GetModuleCfg() *config.ModuleConfig {
	return bc.moduleCfg
}

// StartEventLoop 启动EventLoop循环
func (bc *BaseClient) StartEventLoop() {
	go bc.ProcTopicMsg()
	go bc.ProcWorkerMsg()

	err := bc.subSysParamUpdateEvent()
	if err != nil {
		panic(err.Error())
	}
	err = bc.subNewBlock()
	if err != nil {
		panic(err.Error())
	}
}

// Close 关闭
func (bc *BaseClient) Close() {
	atomic.StoreInt32(&bc.isClosed, 1)
	bc.curCon.Shutdown()
	bc.workerClient.Close()
	bc.topicClient.Close()
	bc.log.Info("consensus base closed")
}

// IsClosed 是否已经关闭
func (bc *BaseClient) IsClosed() bool {
	return atomic.LoadInt32(&bc.isClosed) == 1
}

// Lock 上锁
func (bc *BaseClient) Lock() {
	bc.muLock.Lock()
}

// Unlock 解锁
func (bc *BaseClient) Unlock() {
	bc.muLock.Unlock()
}

// SetLog 设置baseclient的log
func (bc *BaseClient) SetLog(log logging.Log) {
	bc.log = log
}

// SubExecutedBlockFunc 绑定收到执行后区块的回调
func (bc *BaseClient) SubExecutedBlockFunc(ebFunc func(block *pbledger.ExecutedBlock) error) {
	bc.blockExecutedFunc = ebFunc
}

// SubCheckedBlockFunc 绑定收到检验后区块的回调
func (bc *BaseClient) SubCheckedBlockFunc(cbFunc func(block *pbledger.Block) error) {
	bc.blockCheckedFunc = cbFunc
}

// SubReceivedBlockFunc 绑定收到新区快的回调
func (bc *BaseClient) SubReceivedBlockFunc(rbFunc func(block *pbledger.Block) error) {
	bc.blockReceivedFunc = rbFunc
}

// SubReplyAddBlockFunc 绑定添加新区块响应回调
func (bc *BaseClient) SubReplyAddBlockFunc(arFunc func(block *pbledger.Block) error) {
	bc.blockAddReplyFunc = arFunc
}

// SubRelayMsgRequestFunc 绑定中继请求消息回调
func (bc *BaseClient) SubRelayMsgRequestFunc(rcFunc func(msg *p2p.P2PMsg) error) {
	bc.relaymsgRequestFunc = rcFunc
}

// SubRelayMsgResponseFunc 绑定中继响应消息回调
func (bc *BaseClient) SubRelayMsgResponseFunc(rcFunc func(msg *p2p.P2PMsg) error) {
	bc.relayMsgResponseFunc = rcFunc
}

// ProcTopicMsg 处理topic类型消息
func (bc *BaseClient) ProcTopicMsg() {
	// 监听共识模块的topic消息
	for msg := range bc.topicClient.Recv() {
		bc.log.Debugf("consensus topic client receive msg %s[%d]", types.GetEventName(int(msg.Ty)), msg.Ty)
		switch msg.Ty {
		case types.EventAddBlock:
			go mq.ProcessMsg(msg, bc.SetCurrentBlock, nil, bc.log)
		case types.EventDelBlock:
			go mq.ProcessMsg(msg, bc.UpdateCurrentBlock, nil, bc.log)
		case types.EventExecutedBlock:
			go mq.ProcessMsg(msg, bc.ProcessExecutedBlock, nil, bc.log)
		case types.EventCheckedBlock:
			go mq.ProcessMsg(msg, bc.ProcessCheckedBlock, nil, bc.log)
		case types.EventReceivedBlock:
			go mq.ProcessMsg(msg, bc.ProcessReceivedBlock, nil, bc.log)
		case types.EventReplyAddBlock:
			go mq.ProcessMsg(msg, bc.ProcessReplyAddBlock, nil, bc.log)
		case types.EventConsensusRequestMsg:
			go mq.ProcessMsg(msg, bc.ProcessRelayMsgRequest, nil, bc.log)
		case types.EventConsensusResponseMsg:
			go mq.ProcessMsg(msg, bc.ProcessRelayMsgResponse, nil, bc.log)
		case pubsub.SubBlockHeader:
			go mq.ProcessMsg(msg, bc.ProcessSubBlockHeader, nil, bc.log)
		default:
			bc.log.Warnf("Unsupported message type: %s", types.GetEventName(int(msg.Ty)))
		}
	}
}

// ProcWorkerMsg 处理worker类型消息
func (bc *BaseClient) ProcWorkerMsg() {
	// 监听共识模块的topic消息
	for msg := range bc.workerClient.Recv() {
		bc.log.Debugf("consensus worker client receive msg %s[%d]", types.GetEventName(int(msg.Ty)), msg.Ty)
		switch msg.Ty {
		case pubsub.SubSysParam:
			go mq.ProcessMsg(msg, bc.ProcSubSysParam, nil, bc.log)
		case types.EventConsensusQuery:
			go mq.ProcessMsg(msg, bc.ProcessConsensusQuery, nil, bc.log)
		default:
			bc.log.Warnf("Unsupported message type: %s", types.GetEventName(int(msg.Ty)))
		}
	}
}

// IsSynched 是否追上最新高度
func (bc *BaseClient) IsSynched() bool {
	if bc.GetWorkerClient() == nil {
		panic("bc not bind message queue.")
	}
	msg := bc.GetWorkerClient().NewMessage(types.BlockChainModule, mq.ModeWorker.String(), types.EventIsSynched, nil)
	err := bc.GetWorkerClient().Send(msg, true)
	if err != nil {
		return false
	}
	resp, err := bc.GetWorkerClient().Wait(msg)
	if err != nil {
		return false
	}
	return resp.Data.(bool)
}

// RequestBlockHeight 请求当前区块高度
func (bc *BaseClient) RequestBlockHeight() (uint64, error) {
	if bc.GetWorkerClient() == nil {
		panic("bc not bind message queue.")
	}

	msg := bc.GetWorkerClient().NewMessage(types.BlockChainModule, mq.ModeWorker.String(), types.EventGetBlockNumber, nil)
	err := bc.GetWorkerClient().Send(msg, true)
	if err != nil {
		return 0, err
	}
	resp, err := bc.GetWorkerClient().Wait(msg)
	if err != nil {
		return 0, err
	}
	blkHeight := resp.Data.(uint64)
	return blkHeight, nil
}

// RequestBlockSequence 请求blocksequence
func (bc *BaseClient) RequestBlockSequence() (uint64, error) {
	if bc.GetWorkerClient() == nil {
		panic("bc not bind message queue.")
	}

	msg := bc.GetWorkerClient().NewMessage(types.BlockChainModule, mq.ModeWorker.String(), types.EventGetBlockSequence, nil)
	err := bc.GetWorkerClient().Send(msg, true)
	if err != nil {
		return 0, err
	}
	resp, err := bc.GetWorkerClient().Wait(msg)
	if err != nil {
		return 0, err
	}
	blkSeq := resp.Data.(uint64)
	return blkSeq, nil
}

// RequestBlock 查询指定高度的区块
func (bc *BaseClient) RequestBlock(height uint64) (*pbledger.Block, error) {
	if bc.GetWorkerClient() == nil {
		panic("bc not bind message queue.")
	}
	reqblock := &commonproto.ReqBlock{Height: height}
	msg := bc.GetWorkerClient().NewMessage(types.BlockChainModule, mq.ModeWorker.String(), types.EventGetBlock, reqblock)
	err := bc.GetWorkerClient().Send(msg, true)
	if err != nil {
		return nil, err
	}
	resp, err := bc.GetWorkerClient().Wait(msg)
	if err != nil {
		return nil, err
	}
	block := resp.Data.(*pbledger.Block)
	return block, nil
}

// RequestBlocks 从blockchain模块批量请求区块
func (bc *BaseClient) RequestBlocks(start, end uint64) ([]*pbledger.Block, error) {
	if bc.workerClient == nil {
		panic("bc not bind message queue.")
	}
	reqblock := &commonproto.ReqBlocks{Start: start, End: end}
	msg := bc.workerClient.NewMessage(types.BlockChainModule, mq.ModeWorker.String(), types.EventGetBlocks, reqblock)
	err := bc.workerClient.Send(msg, true)
	if err != nil {
		return nil, err
	}

	resp, err := bc.workerClient.WaitTimeout(msg, time.Second*consensustypes.DefaultMqTimeOut)
	if err != nil {
		return nil, err
	}
	blocks := resp.Data.([]*pbledger.Block)
	return blocks, nil
}

// RequestPendingBlocks 从blockchain缓存中拉取待验证区块
func (bc *BaseClient) RequestPendingBlocks(start, end uint64) ([]*pbledger.Block, error) {
	if bc.workerClient == nil {
		panic("bc not bind message queue.")
	}
	reqblock := &commonproto.ReqBlocks{Start: start, End: end}
	msg := bc.workerClient.NewMessage(types.BlockChainModule, mq.ModeWorker.String(), types.EventRequestBlock, reqblock)
	err := bc.workerClient.Send(msg, true)
	if err != nil {
		return nil, err
	}
	resp, err := bc.workerClient.Wait(msg)
	if err != nil {
		return nil, err
	}
	blocks := resp.Data.([]*pbledger.Block)
	return blocks, nil
}

// RequestSequenceCenterBlocks 批量查询sequenceCenter中区块
func (bc *BaseClient) RequestSequenceCenterBlocks(start, end uint64) ([]*pbledger.Block, error) {
	if bc.workerClient != nil {
		panic("Worker client message queue not initialized.")
	}

	req := commonproto.ReqBlocks{Start: start, End: end}
	msg := bc.workerClient.NewMessage(types.BlockChainModule, mq.ModeWorker.String(), types.EventQuerySequenceCenterBlock, req)

	err := bc.workerClient.Send(msg, true)
	if err != nil {
		return nil, err
	}

	resp, err := bc.workerClient.WaitTimeout(msg, time.Second*consensustypes.DefaultMqTimeOut)
	if err != nil {
		return nil, err
	}

	blocks := resp.Data.([]*pbledger.Block)
	return blocks, nil
}

// RequestLastBlock 获取最新的block从blockchain模块
func (bc *BaseClient) RequestLastBlock() (*pbledger.Block, error) {
	if bc.workerClient == nil {
		panic("client not bind message queue.")
	}
	msg := bc.workerClient.NewMessage(types.BlockChainModule, mq.ModeWorker.String(), types.EventGetLastBlock, nil)
	err := bc.workerClient.Send(msg, true)
	if err != nil {
		return nil, err
	}
	resp, err := bc.workerClient.Wait(msg)
	if err != nil {
		return nil, err
	}
	eb := resp.Data.(*pbledger.ExecutedBlock)
	return eb.Block, nil
}

// RequestLastExecutedBlock 查询最新执行区块
func (bc *BaseClient) RequestLastExecutedBlock() (*pbledger.ExecutedBlock, error) {
	if bc.workerClient == nil {
		panic("client not bind message queue.")
	}
	msg := bc.workerClient.NewMessage(types.BlockChainModule, mq.ModeWorker.String(), types.EventGetLastExecutedBlock, nil)
	err := bc.workerClient.Send(msg, true)
	if err != nil {
		return nil, err
	}
	resp, err := bc.workerClient.Wait(msg)
	if err != nil {
		return nil, err
	}
	block := resp.Data.(*pbledger.ExecutedBlock)
	return block, nil
}

// RequestExecutedBlock 查询指定高度的执行区块
func (bc *BaseClient) RequestExecutedBlock(height uint64) (*pbledger.ExecutedBlock, error) {
	if bc.workerClient == nil {
		panic("client not bind message queue.")
	}
	var req = &commonproto.ReqBlock{Height: height}
	msg := bc.workerClient.NewMessage(types.BlockChainModule, mq.ModeWorker.String(), types.EventGetExecutedBlock, req)
	err := bc.workerClient.Send(msg, true)
	if err != nil {
		return nil, err
	}
	resp, err := bc.workerClient.WaitTimeout(msg, time.Second*consensustypes.DefaultMqTimeOut)
	if err != nil {
		return nil, err
	}
	eb := resp.Data.(*pbledger.ExecutedBlock)
	return eb, nil
}

// RequestExecutedBlocks 批量查询执行过的区块
func (bc *BaseClient) RequestExecutedBlocks(start, end uint64) ([]*pbledger.ExecutedBlock, error) {
	if bc.workerClient == nil {
		panic("client not bind message queue.")
	}

	var req = &commonproto.ReqBlocks{Start: start, End: end}
	msg := bc.workerClient.NewMessage(types.BlockChainModule, mq.ModeWorker.String(), types.EventGetExecutedBlocks, req)
	err := bc.workerClient.Send(msg, true)
	if err != nil {
		return nil, err
	}
	resp, err := bc.workerClient.Wait(msg)
	if err != nil {
		return nil, err
	}
	ebs := resp.Data.([]*pbledger.ExecutedBlock)
	return ebs, nil
}

// RequestMaxExecutedBlockNumber 查询最大执行区块高度
func (bc *BaseClient) RequestMaxExecutedBlockNumber() (uint64, error) {
	if bc.workerClient == nil {
		panic("client not bind message queue.")
	}
	reqMsg := bc.GetWorkerClient().NewMessage(types.BlockChainModule, mq.ModeWorker.String(), types.EventGetMaxExecutedBlockNumber, nil)
	err := bc.GetWorkerClient().Send(reqMsg, true)
	if err != nil {
		bc.log.Errorf("Send EventGetMaxExecutedBlockNumber request failed, err:%s\n", err.Error())
		return 0, nil
	}

	resp, err := bc.GetWorkerClient().WaitTimeout(reqMsg, time.Second*consensustypes.DefaultMqTimeOut)
	if err != nil {
		bc.log.Errorf("Wait for EventGetMaxExecutedBlockNumber response failed, err:%s\n", err.Error())
		return 0, nil
	}

	maxExecutedBlockHeight := resp.Data.(uint64)
	return maxExecutedBlockHeight, nil
}

// RequestMaxExecutedBlock 查询最大执行过的区块
func (bc *BaseClient) RequestMaxExecutedBlock() (*pbledger.ExecutedBlock, error) {
	if bc.workerClient == nil {
		panic("client not bind message queue.")
	}
	reqMsg := bc.GetWorkerClient().NewMessage(types.BlockChainModule, mq.ModeWorker.String(), types.EventGetLastBlock, nil)
	err := bc.GetWorkerClient().Send(reqMsg, true)
	if err != nil {
		bc.log.Errorf("Send EventGetMaxExecutedBlockNumber request failed, err:%s\n", err.Error())
		return nil, err
	}

	resp, err := bc.GetWorkerClient().WaitTimeout(reqMsg, time.Second*consensustypes.DefaultMqTimeOut)
	if err != nil {
		bc.log.Errorf("Wait for EventGetMaxExecutedBlockNumber response failed, err:%s\n", err.Error())
		return nil, err
	}

	eb := resp.Data.(*pbledger.ExecutedBlock)
	return eb, nil
}

// PreRemoveTransactions 从PendingPool中取交易
func (bc *BaseClient) PreRemoveTransactions() ([]*pbledger.Transaction, error) {
	if bc.workerClient == nil {
		panic("bc not bind message queue.")
	}
	reqTxSize := bc.GetCurParam().GetBlockLimit().GetMaxTxCount()
	if reqTxSize <= 0 {
		return nil, types.ErrInvalidParam
	}
	msg := bc.workerClient.NewMessage(types.P2pModule, mq.ModeWorker.String(), types.EventTxPoolPreRemove, int(reqTxSize))
	err := bc.workerClient.Send(msg, true)
	if err != nil {
		return nil, err
	}
	resp, err := bc.workerClient.WaitTimeout(msg, time.Second*consensustypes.DefaultMqTimeOut)
	if err != nil {
		return nil, err
	}
	return resp.Data.([]*pbledger.Transaction), nil
}

// PreRemoveSpecialTransactions 将特定交易标记为“待删除”（返回值[]int：影响记录的条数）
func (bc *BaseClient) PreRemoveSpecialTransactions(txhashes []protos.Hash) error {
	msg := bc.topicClient.NewMessage(types.P2pModule, mq.ModeWorker.String(), types.EventTxPoolPreRemoveByHash, txhashes)
	err := bc.topicClient.Send(msg, false)
	if err != nil {
		return err
	}

	return nil
}

// RemoveTransactions 从缓冲池中批量移除交易
func (bc *BaseClient) RemoveTransactions(txs []protos.Hash) error {
	msg := bc.topicClient.NewMessage(types.P2pModule, mq.ModeWorker.String(), types.EventTxPoolRemoveTxs, txs)
	err := bc.topicClient.Send(msg, false)
	if err != nil {
		return err
	}

	return nil
}

// UndoPreRemoveTransactions 去掉交易上的“待删除”标记（返回值[]int：影响记录的条数）
func (bc *BaseClient) UndoPreRemoveTransactions(txhashes []protos.Hash) ([]int64, error) {
	msg := bc.workerClient.NewMessage(types.P2pModule, mq.ModeWorker.String(), types.EventTxPoolUndoPreRemove, txhashes)
	err := bc.workerClient.Send(msg, true)
	if err != nil {
		return nil, err
	}
	resp, err := bc.workerClient.WaitTimeout(msg, time.Second*consensustypes.DefaultMqTimeOut)
	if err != nil {
		return nil, err
	}
	return resp.Data.([]int64), nil
}

// WriteBlock 共识生成区块之后，发送到blockchain进行后续操作
func (bc *BaseClient) WriteBlock(block *pbledger.Block) error {
	//保存block的原始信息用于删除mempool中的错误交易

	msg := bc.GetTopicClient().NewMessage(types.BlockChainModule, mq.ModeTopic.String(), types.EventAddBlock, block)
	err := bc.GetTopicClient().Send(msg, false)
	if err != nil {
		return err
	}

	return nil
}

// SaveBlock 保存区块
func (bc *BaseClient) SaveBlock(blockNum uint64) error {
	msg := bc.topicClient.NewMessage(types.BlockChainModule, mq.ModeTopic.String(), types.EventSaveBlock, blockNum)
	err := bc.topicClient.Send(msg, true)
	if err != nil {
		return err
	}

	return nil
}

// SetCurrentBlock 设置当前区块
func (bc *BaseClient) SetCurrentBlock(msg *mq.Message) {
	block := msg.Data.(*pbledger.Block)
	bc.muLock.Lock()
	bc.currentBlock = block
	bc.muLock.Unlock()
}

// UpdateCurrentBlock 更新当前区块
func (bc *BaseClient) UpdateCurrentBlock(msg *mq.Message) {
	block := msg.Data.(*pbledger.Block)
	bc.muLock.Lock()
	defer bc.muLock.Unlock()
	block, err := bc.RequestLastBlock()
	if err != nil {
		bc.log.Error("UpdateCurrentBlock", "RequestLastBlock", err)
		return
	}
	bc.currentBlock = block
}

// WriteSequenceCenterBlock 存储序列化中心区块
func (bc *BaseClient) WriteSequenceCenterBlock(block *pbledger.Block) error {
	if bc.workerClient == nil {
		panic("client not bind message queue.")
	}

	msg := bc.workerClient.NewMessage(types.BlockChainModule, mq.ModeWorker.String(), types.EventAddSequenceCenterBlock, block)
	err := bc.workerClient.Send(msg, true)
	if err != nil {
		return err
	}

	return nil
}

// ProcessExecutedBlock 处理执行后的区块
func (bc *BaseClient) ProcessExecutedBlock(msg *mq.Message) {
	eb := msg.Data.(*pbledger.ExecutedBlock)
	if bc.blockExecutedFunc != nil {
		err := bc.blockExecutedFunc(eb)
		if err != nil {
			bc.log.Errorf("ProcessExecutedBlock failed, err: %s", err.Error())
		}
	}
}

// ProcessCheckedBlock 处理校验后的区块
func (bc *BaseClient) ProcessCheckedBlock(msg *mq.Message) {
	blk := msg.Data.(*pbledger.Block)
	if bc.blockCheckedFunc != nil {
		err := bc.blockCheckedFunc(blk)
		if err != nil {
			bc.log.Errorf("ProcessCheckedBlock failed, err: %s", err.Error())
		}
	}
}

// ProcessReceivedBlock 处理收到的区块
func (bc *BaseClient) ProcessReceivedBlock(msg *mq.Message) {
	blk := msg.Data.(*pbledger.Block)
	if bc.blockReceivedFunc != nil {
		err := bc.blockReceivedFunc(blk)
		if err != nil {
			bc.log.Errorf("ProcessReceivedBlock failed, err: %s", err.Error())
		}
	}
}

// ProcessReplyAddBlock 处理通过P2P转发的区块
func (bc *BaseClient) ProcessReplyAddBlock(msg *mq.Message) {
	block := msg.Data.(*pbledger.Block)
	if bc.blockAddReplyFunc != nil {
		err := bc.blockAddReplyFunc(block)
		if err != nil {
			bc.log.Errorf("ProcessReplyAddBlock failed, err: %s", err.Error())
		}
	}
}

// ProcessRelayMsgRequest 处理通过P2P转发的请求
func (bc *BaseClient) ProcessRelayMsgRequest(msg *mq.Message) {
	req := msg.Data.(*p2p.P2PMsg)
	if bc.relaymsgRequestFunc != nil {
		err := bc.relaymsgRequestFunc(req)
		if err != nil {
			bc.log.Errorf("ProcessRelayMsgRequest failed, err: %s", err.Error())
		}
	}
}

// ProcessRelayMsgResponse 处理通过P2P转发的响应
func (bc *BaseClient) ProcessRelayMsgResponse(msg *mq.Message) {
	response := msg.Data.(*p2p.P2PMsg)
	if bc.relayMsgResponseFunc != nil {
		err := bc.relayMsgResponseFunc(response)
		if err != nil {
			bc.log.Errorf("ProcessRelayMsgResponse failed, err: %s", err.Error())
		}
	}
}

// GetCurrentBlock 获取当前区块
func (bc *BaseClient) GetCurrentBlock() (b *pbledger.Block) {
	bc.muLock.Lock()
	defer bc.muLock.Unlock()
	return bc.currentBlock
}

// GetCurrentHeight 获取当前高度
func (bc *BaseClient) GetCurrentHeight() uint64 {
	bc.muLock.Lock()
	defer bc.muLock.Unlock()
	start := bc.currentBlock.Header.Number
	return start
}

// RequestCurrentHeight 获取当前高度
func (bc *BaseClient) RequestCurrentHeight() (uint64, error) {
	return 0, nil
}

// ReqBlockByHash 通过区块hash获取区块信息
func (bc *BaseClient) ReqBlockByHash(hash []byte) (*pbledger.ExecutedBlock, error) {
	if bc.topicClient == nil {
		panic("bc not bind message queue.")
	}
	reqHash := commonproto.ReqHash{Hash: hash}
	msg := bc.topicClient.NewMessage(types.BlockChainModule, mq.ModeTopic.String(), types.EventGetBlockByHash, &reqHash)
	err := bc.topicClient.Send(msg, true)
	if err != nil {
		return nil, err
	}
	resp, err := bc.topicClient.Wait(msg)
	if err != nil {
		return nil, err
	}
	block := resp.Data.(*pbledger.ExecutedBlock)
	return block, nil
}

// ReqBlocksByHashes 通过区块[]hash获取区块数组信息
func (bc *BaseClient) ReqBlocksByHashes(hashes [][]byte) ([]*pbledger.ExecutedBlock, error) {
	if bc.topicClient == nil {
		panic("bc not bind message queue.")
	}
	reqHashes := commonproto.ReqHashes{Hashes: hashes}
	msg := bc.topicClient.NewMessage(types.BlockChainModule, mq.ModeTopic.String(), types.EventGetBlocksByHashes, &reqHashes)
	err := bc.topicClient.Send(msg, true)
	if err != nil {
		return nil, err
	}
	resp, err := bc.topicClient.Wait(msg)
	if err != nil {
		return nil, err
	}
	blocks := resp.Data.([]*pbledger.ExecutedBlock)
	return blocks, nil
}

// BroadCastBlock 广播区块
func (bc *BaseClient) BroadCastBlock(block *pbledger.Block, broadcastTimeOut int64) error {
	if bc.topicClient == nil {
		panic("bc not bind message queue.")
	}

	msg := bc.topicClient.NewMessage(types.P2pModule, mq.ModeTopic.String(), types.EventBlockBroadcast, block)
	err := bc.topicClient.Send(msg, false)
	if err != nil {
		return err
	}

	return nil
}

// StoreElectionContextInfo 存储投票信息
func (bc *BaseClient) StoreElectionContextInfo(electionContextBytes []byte) error {
	if bc.topicClient == nil {
		panic("bc not bind message queue.")
	}

	msg := bc.topicClient.NewMessage(types.BlockChainModule, mq.ModeTopic.String(), types.EventStoreElectionContext, electionContextBytes)
	err := bc.topicClient.Send(msg, false)
	if err != nil {
		return err
	}

	return nil
}

// RequestElectionContextInfo 请求投票信息
func (bc *BaseClient) RequestElectionContextInfo() ([]byte, error) {
	if bc.workerClient == nil {
		panic("bc not bind message queue.")
	}

	msg := bc.workerClient.NewMessage(types.BlockChainModule, mq.ModeWorker.String(), types.EventRequestElectionContext, nil)
	err := bc.workerClient.Send(msg, true)
	if err != nil {
		return nil, err
	}
	resp, err := bc.workerClient.WaitTimeout(msg, time.Second*consensustypes.DefaultMqTimeOut)
	if err != nil {
		return nil, err
	}

	replyElectionBytes := resp.Data.([]byte)
	return replyElectionBytes, nil
}

// StoreRoundMinersInfo 存储轮次Miner信息
func (bc *BaseClient) StoreRoundMinersInfo(miners *dpos.RoundMiners) error {
	if bc.topicClient == nil {
		panic("bc not bind message queue.")
	}

	roundMinersBytes, err := proto.Marshal(miners)
	if err != nil {
		bc.log.Errorf("Marshal round miners info failed, err:%s\n", err.Error())
		return err
	}

	msg := bc.topicClient.NewMessage(types.BlockChainModule, mq.ModeTopic.String(), types.EventStoreRoundMinersInfo, roundMinersBytes)
	err = bc.topicClient.Send(msg, true)
	if err != nil {
		return err
	}

	return nil
}

// RequestRoundMinersInfo 请求轮次Miner信息
func (bc *BaseClient) RequestRoundMinersInfo(reqRound int64, requestTimeOut uint64) ([]byte, error) {
	if bc.workerClient == nil {
		panic("bc not bind message queue.")
	}

	msg := bc.workerClient.NewMessage(types.BlockChainModule, mq.ModeWorker.String(), types.EventRequestRoundMinersInfo, int(reqRound))
	err := bc.workerClient.Send(msg, true)
	if err != nil {
		return nil, err
	}
	resp, err := bc.workerClient.WaitTimeout(msg, time.Duration(requestTimeOut)*time.Second)
	if err != nil {
		return nil, err
	}

	roundMinerInfo := resp.Data.([]byte)
	return roundMinerInfo, nil
}

// StoreTopicInfo 存储Topic信息
func (bc *BaseClient) StoreTopicInfo(topicInfo []byte, topicType int64) error {
	if bc.GetTopicClient() == nil {
		panic("bc not bind message queue.")
	}

	msg := bc.GetTopicClient().NewMessage(types.BlockChainModule, mq.ModeTopic.String(), topicType, topicInfo)
	err := bc.GetTopicClient().Send(msg, true)
	if err != nil {
		return err
	}
	return nil
}

// RequestTopicInfo 请求Topic信息
func (bc *BaseClient) RequestTopicInfo(topicType int64) ([]byte, error) {
	if bc.workerClient == nil {
		panic("bc not bind message queue.")
	}

	msg := bc.workerClient.NewMessage(types.BlockChainModule, mq.ModeWorker.String(), topicType, nil)
	err := bc.workerClient.Send(msg, true)
	if err != nil {
		return nil, err
	}

	resp, err := bc.workerClient.WaitTimeout(msg, time.Second*consensustypes.DefaultMqTimeOut)
	if err != nil {
		return nil, err
	}

	topicInfo := resp.Data.([]byte)
	return topicInfo, nil
}

// RequestConsensusStepRecords 请求共识切换记录
func (bc *BaseClient) RequestConsensusStepRecords() (*pbledger.ConsensusStepRecords, error) {
	return nil, nil
}

// RequestSequenceMaxHeight 请求区块最大序列号
func (bc *BaseClient) RequestSequenceMaxHeight() (uint64, error) {
	if bc.workerClient == nil {
		panic("bc not bind message queue.")
	}

	msg := bc.workerClient.NewMessage(types.BlockChainModule, mq.ModeWorker.String(), types.EventQueryCurrentBlockSequence, nil)
	err := bc.workerClient.Send(msg, true)
	if err != nil {
		return 0, err
	}
	resp, err := bc.workerClient.WaitTimeout(msg, time.Second*consensustypes.DefaultMqTimeOut)
	if err != nil {
		return 0, err
	}

	height := resp.Data.(uint64)
	return height, nil
}

// SendRelayMsg 发送Relay消息
func (bc *BaseClient) SendRelayMsg(msg *p2p.P2PMsg) error {
	newMsg := bc.GetTopicClient().NewMessage(module.P2pModuleName, mq.ModeTopic.String(), types.EventConsensusRequestMsg, msg)
	err := bc.GetTopicClient().Send(newMsg, false)
	if err != nil {
		return err
	}
	return nil
}

// InitLastBlockMessage 初始化lastBlockMessage
func (bc *BaseClient) InitLastBlockMessage() {
	if bc.lastBlockCheckedMessage == nil {
		bc.lastBlockCheckedMessage = &atomic.Value{}
	}
	if bc.lastBlockSequence == nil {
		bc.lastBlockSequence = &atomic.Value{}
	}
}

// SetLastBlockCheckedMessage 设置最新校验区块
func (bc *BaseClient) SetLastBlockCheckedMessage(lastBlockCheckedMsg *commonproto.BlockCheckedMessage) {
	curCheckedBlkNum := bc.GetLastBlockCheckedMessage().BlockNumber
	lastCheckedBlkNum := lastBlockCheckedMsg.BlockNumber
	if bc.GetLastBlockCheckedMessage().BlockHash != nil && curCheckedBlkNum >= lastCheckedBlkNum {
		bc.log.Debugf("no need to update last checked message, current %d and dest %d", curCheckedBlkNum, lastCheckedBlkNum)
		return
	}

	bc.lastBlockCheckedMessage.Store(lastBlockCheckedMsg)

	if bc.lastBlockSequence.Load() == nil || (bc.lastBlockSequence.Load() != nil && lastBlockCheckedMsg.BlockSequence > bc.lastBlockSequence.Load().(uint64)) {
		bc.lastBlockSequence.Store(lastBlockCheckedMsg.BlockSequence)
	}
}

// SetLastBlockSequence 设置last blocksequence
func (bc *BaseClient) SetLastBlockSequence(seq uint64) {
	bc.lastBlockSequence.Store(seq)
}

// UpdateLastBlockCheckedMessage 更新最新校验区块
func (bc *BaseClient) UpdateLastBlockCheckedMessage(blk *pbledger.Block) {
	bc.log.Debugf("blk Number %d and last block checked %d", blk.Header.Number, bc.GetLastBlockCheckedMessage().BlockNumber)
	if blk.Header.Number > bc.GetLastBlockCheckedMessage().BlockNumber {
		var lastBlkCheckMsg = &commonproto.BlockCheckedMessage{}
		lastBlkCheckMsg.BlockNumber = blk.Header.Number
		if blk.Header.BlockSequence == 0 {
			lastBlkCheckMsg.BlockSequence = bc.GetLastBlockCheckedMessage().BlockSequence
		} else {
			lastBlkCheckMsg.BlockSequence = blk.Header.BlockSequence
		}
		lastBlkCheckMsg.BlockHash = blk.Hash()
		lastBlkCheckMsg.BlockTime = blk.Header.BlockTime
		bc.lastBlockCheckedMessage.Store(lastBlkCheckMsg)
		bc.lastBlockSequence.Store(lastBlkCheckMsg.GetBlockSequence())
		bc.RemovePreRemovedTxHashesFromLocalCache(lastBlkCheckMsg.BlockHash)
		bc.log.Infof("Adopt as last checked block %d and seq %d", lastBlkCheckMsg.BlockNumber, lastBlkCheckMsg.BlockSequence)
	}
}

// GetLastBlockCheckedMessage 获取最新校验区块信息
func (bc *BaseClient) GetLastBlockCheckedMessage() *commonproto.BlockCheckedMessage {
	if bc.lastBlockCheckedMessage != nil {
		if blkCheckedMsg, ok := bc.lastBlockCheckedMessage.Load().(*commonproto.BlockCheckedMessage); ok {
			return blkCheckedMsg
		}
	}
	return &commonproto.BlockCheckedMessage{}
}

// GetLastBlockSequence 获取最新block sequence
func (bc *BaseClient) GetLastBlockSequence() uint64 {
	if bc.lastBlockSequence != nil {
		return bc.lastBlockSequence.Load().(uint64)
	}
	return 0
}

// PutPreRemovedTxHashesIntoLocalCache 将预删除交易的哈希缓存在本地，miner出块或背书成功时会调用此方法，
// 记录预删除交易。如果缓存中的条目过期了，则需要取消交易的预删除标记
func (bc *BaseClient) PutPreRemovedTxHashesIntoLocalCache(blockHash protos.Hash, txHashes []protos.Hash, log logging.Log) {
	blockHashHex := hex.EncodeToString(blockHash)
	log.Info("PutPreRemovedTxHashesIntoLocalCache")
	err := bc.markedTxLocalCache.Set(blockHashHex, util.TxHashesToBytes(txHashes))
	if err != nil {
		log.Warnf("PutPreRemovedTxHashesIntoLocalCache failed:%s", err.Error())
	}
}

// RemovePreRemovedTxHashesFromLocalCache 交易出块后调用此方法删除缓存的指定块中的交易哈希
func (bc *BaseClient) RemovePreRemovedTxHashesFromLocalCache(blockHash protos.Hash) {
	blockHashHex := hex.EncodeToString(blockHash)
	blkLog := bc.log.WithField(logging.LogFieldBlockHash, blockHashHex[0:8])
	blkLog.Info("RemovePreRemovedTxHashesFromLocalCache")
	if bc.markedTxLocalCache != nil {
		err := bc.markedTxLocalCache.Delete(blockHashHex)
		if err != nil {
			blkLog.Warnf("RemovePreRemovedTxHashesFromLocalCache failed:%s", err.Error())
		}
	}
}

// SetMarkedTxLocalCache 设置本地标记区块缓存
func (bc *BaseClient) SetMarkedTxLocalCache(cache *bigcache.BigCache) {
	bc.markedTxLocalCache = cache
}

// GetBlockExecutedMessage 获取区块执行完毕消息
func (bc *BaseClient) GetBlockExecutedMessage(blockNumber uint64, log logging.Log) *commonproto.BlockExecutedMessage {
	bc.checkInitialized()
	var msg *commonproto.BlockExecutedMessage
	msgCacheKey := strconv.FormatUint(blockNumber, 10)
	if blkExecuted, ok := bc.blockExecutedMessage.Get(msgCacheKey); ok {
		return blkExecuted.(*commonproto.BlockExecutedMessage)
	}

	//double check
	if ebMsg, ok := bc.blockExecutedMessage.Get(msgCacheKey); ok {
		return ebMsg.(*commonproto.BlockExecutedMessage)
	}

	log.Debug("fetching exec result from ledger...")
	//block, err := service.GetBlock(blockNumber)
	blockReq := &commonproto.ReqBlock{Height: blockNumber}
	req := bc.workerClient.NewMessage(types.BlockChainModule, mq.ModeWorker.String(), types.EventGetExecutedBlock, blockReq)
	err := bc.workerClient.Send(req, true)
	if err != nil {
		log.Warnf("Send block request to block chain failed, err: %s", err.Error())
		return nil
	}

	var resp = &mq.Message{}
	resp, err = bc.workerClient.WaitTimeout(req, time.Second*consensustypes.DefaultMqTimeOut)
	if err != nil {
		log.Warnf("Wait response from client failed, err :%s", err.Error())
		return nil
	}

	block := resp.Data.(*pbledger.ExecutedBlock)
	if block == nil {
		reply := resp.Data.(*commonproto.Reply)
		log.Debug(" Fetch block failed, err :%s\n", string(reply.Msg))
		return nil
	}

	log.Debug("got exec result from ledger")
	msg = &commonproto.BlockExecutedMessage{
		BlockNumber: block.Block.Header.Number,
		BlockHash:   block.Block.Hash(),
		StateHash:   block.BlockExecResult.StateHash,
		BlockTime:   block.Block.Header.BlockTime,
	}

	bc.blockExecutedMessage.Set(msgCacheKey, msg)
	log.Debug("exec result cache by GetBlockExecutedMessage")
	return msg
}

func (bc *BaseClient) checkInitialized() {
	if bc.blockExecutedMessage == nil {
		bc.blockExecutedMessage = cache.New()
	}
}

// PutBlockExecutedMessage 将区块执行完毕消息放到队列中
func (bc *BaseClient) PutBlockExecutedMessage(msg *commonproto.BlockExecutedMessage, log logging.Log) {
	bc.checkInitialized()
	msgCacheKey := strconv.FormatUint(msg.BlockNumber, 10)
	if _, ok := bc.blockExecutedMessage.Get(msgCacheKey); ok {
		log.Debug("exists in cache")
		return
	}

	bc.blockExecutedMessage.Set(msgCacheKey, msg)
	log.Debugf("cached by PutBlockExecutedMessage,TransactionExecResultsHash:%s", hex.EncodeToString(msg.StateHash))
}

// InitBlockSynchronized 初始化BlockSynchronized
func (bc *BaseClient) InitBlockSynchronized() {
	if bc.blockSynchronized == nil {
		bc.blockSynchronized = &atomic.Value{}
	}
}

// SetBlockSynchronized 设置区块同步状态
func (bc *BaseClient) SetBlockSynchronized(bVal bool) {
	if bc.blockSynchronized != nil {
		bc.blockSynchronized.Store(bVal)
	}
}

// BlockSynchronized 获取区块同步状态
func (bc *BaseClient) BlockSynchronized() bool {
	if bc.blockSynchronized != nil {
		if res, ok := bc.blockSynchronized.Load().(bool); ok {
			return res
		}
	}
	return false
}

// ProcessSubBlockHeader 处理区块头订阅消息
func (bc *BaseClient) ProcessSubBlockHeader(msg *mq.Message) {
	rsp := msg.Data.(*commonproto.RspSubscriber)
	var blkHeader = &pbledger.BlockHeader{}
	err := proto.Unmarshal(rsp.GetNormal().Data, blkHeader)
	if err != nil {
		panic(err.Error())
	}

	height := blkHeader.GetNumber()
	bc.StepParamListCheck(height)
	bc.StepConListCheck(height)
}

// CalcMaxTxsSizeInBlock 计算区块中最大交易字节数
func (bc *BaseClient) CalcMaxTxsSizeInBlock() int {
	maxBlockHeaderSize := consensustypes.EstimateMaxBlockHeaderSizeBytes()
	maxTotalSizeOfTxsInBlock := int(bc.GetCurParam().GetBlockLimit().MaxBlockSizeBytes) - maxBlockHeaderSize*2
	if maxTotalSizeOfTxsInBlock < int(bc.GetCurParam().GetBlockLimit().MaxTxSizeBytes) {
		err := fmt.Errorf("blockLimit.MaxBlockSizeBytes must be greater than %d", maxBlockHeaderSize*2+int(bc.GetCurParam().GetBlockLimit().MaxTxSizeBytes))
		panic(err)
	}
	bc.log.Infof("estimated maxTotalSizeOfTxsInBlock:%d", maxTotalSizeOfTxsInBlock)
	return maxTotalSizeOfTxsInBlock
}

// ProcessConsensusQuery 查询共识相关信息
func (bc *BaseClient) ProcessConsensusQuery(msg *mq.Message) {
	var consensusInfo = &protosconfig.ConsensusParam{}
	if bc.GetCurParam() != nil && bc.GetCurParam().GetConsensus() != nil {
		consensusInfo = bc.GetCurParam().GetConsensus()
	}

	replyMsg := bc.workerClient.NewMessage("", "", msg.Ty, consensusInfo)
	msg.Reply(replyMsg)
}
